package com.ctsi.commons.util.async;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * 
 * @author yueming
 * 
 */
public class TaskManager extends Thread {
	private static final Logger log = LoggerFactory.getLogger(TaskManager.class);

	private volatile boolean terminate = false;

	public static final class IntenalRun implements Runnable {
		private final TaskRun task;

		private final Object[] intenalArgs;
		private final ListenableFuture<Object> fu;

		public IntenalRun(TaskRun task, Object[] taskArgs, ListenableFuture<Object> fu) {
			this.fu = fu;
			this.task = task;
			this.intenalArgs = taskArgs;
		}
        @Override
		public void run() {
			try {
				if (!fu.isCancelled()) {
					Object result = task.run(intenalArgs);
					fu.setSuccess(result);
				} else {
					log.info("task {} canceled", task);
				}

			} catch (Exception e) {
				log.debug("run error", e);
				fu.setFailure(e);
			}

		}

	}

	TaskManager() {

	}

	private static final DelayQueue<Pulse> delayQueue = new DelayQueue<Pulse>();

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.itc.commons.util.async.TaskMan#execute(com.itc.commons.util.async.TaskRun, java.lang.Object[])
	 */
	public ListenableFuture<?> execute(final TaskRun task, final Object[] args, final long waiteTime) {
		if(waiteTime>0){
			final TaskFuture<Object> fu = new TaskFuture<>();
			delayQueue.add(new Pulse(waiteTime) {

				@Override
				public void run() {
					try {
						if (!fu.isCancelled()) {
							final Object result = task.run(args);
							fu.setSuccess(result);
						} else {
							log.info("task {} canceled", task);
						}

					} catch (Exception e) {
						log.debug("run error", e);
						fu.setFailure(e);
					}

				}

			});

			startDelayThread();
			return fu;			
		}else{
			return execute(task,  args);
		}

	}

	private void startDelayThread() {
		if (!delayStart) {
			synchronized (this) {
				if (!delayStart) {
					delayStart = true;

					TaskManagerHolder.get_GLOBAL_RUN().execute(new ExecutionPoolPulseWorker());
					log.info("start delay run thread");
				}

			}

		}
	}

	private volatile boolean delayStart = false;

	class ExecutionPoolPulseWorker implements Runnable {
		@Override
		public void run() {
			try {
				while (!terminate) {
					Pulse p = delayQueue.poll(5, TimeUnit.SECONDS);
					if (p != null) {
						TaskManagerHolder.get_GLOBAL_RUN().execute(p);
						log.info("run task {}", p);
					}

				}
			} catch (InterruptedException e) {
				delayStart = false;
				log.warn("ExecutionPoolPulseWorker error ", e);
			}
		}
	}

	public ListenableFuture<?> execute(final TaskRun task, final Object[] args) {
		final TaskFuture<Object> result = new TaskFuture<>();
		TaskManagerHolder.get_GLOBAL_RUN().execute(new IntenalRun(task, args, result));
		return result;
	}

	public void execute(final TaskRun task, final Object[] args, final ListenableFuture<Object> fu) {

		TaskManagerHolder.get_GLOBAL_RUN().execute(new IntenalRun(task, args, fu));

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.itc.commons.util.async.TaskMan#terminate()
	 */
	public void terminateDelayTask() {


		terminate = true;

		log.info("set terminate  true");

	}
	
	
	public void enableDelayTask() {

 

		terminate = false;

		log.info("enable delay task");

	}

	public static abstract class Pulse implements Delayed, Runnable {
		protected final long expireTimeMillisecond;
		protected final long loadTimeMillisecond;

		protected Pulse(long delayMillisecond) {
			this(System.currentTimeMillis(), delayMillisecond);
		}

		protected Pulse(long loadMillisecond, long delayMillisecond) {
			this.loadTimeMillisecond = loadMillisecond;
			expireTimeMillisecond = loadMillisecond + delayMillisecond;
		}

		public long getLoadTimeNanos() {
			return loadTimeMillisecond;
		}

		public long getExpireTimeNanos() {
			return expireTimeMillisecond;
		}

		@Override
		public final long getDelay(TimeUnit unit) {
			return unit.convert(expireTimeMillisecond - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
		}

		public final int compareTo(Delayed other) {
			long r = (expireTimeMillisecond - ((Pulse) other).expireTimeMillisecond);
			if (r < 0)
				return -1;
			if (r > 0)
				return 1;
			return 0;
		}
	}
}
